package com.muse.springbootdemo.service;

import javax.annotation.Resource;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import com.muse.springbootdemo.entity.Message;

import lombok.extern.slf4j.Slf4j;

/**
 * @description
 * @author: muse
 **/
@Slf4j
@Service
public class ProducerService {
    private final static String TOPIC_NAME = "muse-rp";
    private final static Integer PARTITION_ONE = 0;
    private final static Integer PARTITION_TWO = 1;

    @Resource
    private KafkaTemplate<String, Message> kafkaTemplate;

    /**
     * 同步阻塞——消息发送
     */
    public void blockingSendMsg() throws Throwable {
        Message message;
        for (int i=0; i< 5; i++) {
            message = new Message(String.valueOf(i), "BLOCKING_MSG_SPRINGBOOT_" + i);
            ListenableFuture<SendResult<String, Message>> future = kafkaTemplate.send(TOPIC_NAME, PARTITION_ONE,
                    "" + message.getMegId(), message);
            SendResult<String, Message> sendResult = future.get();
            RecordMetadata recordMetadata = sendResult.getRecordMetadata();
            log.info("---BLOCKING_MSG_SPRINGBOOT--- [topic]={}, [partition]={}, [offset]={}", recordMetadata.topic(),
                    recordMetadata.partition(), recordMetadata.offset());
        }
    }

    /**
     * 异步回调——消息发送
     */
    public void noBlockingSendMsg() {
        Message message;
        for (int i=0; i< 5; i++) {
            message = new Message(String.valueOf(i), "NO_BLOCKING_MSG_SPRINGBOOT_" + i);
            ListenableFuture<SendResult<String, Message>> future = kafkaTemplate.send(TOPIC_NAME, PARTITION_ONE,
                    "" + message.getMegId(), message);
            future.addCallback(new ListenableFutureCallback<SendResult<String, Message>>() {
                @Override
                public void onFailure(Throwable ex) {
                    log.error("消息发送失败！", ex);
                }

                @Override
                public void onSuccess(SendResult<String, Message> result) {
                    RecordMetadata recordMetadata = result.getRecordMetadata();
                    log.info("---NO_BLOCKING_MSG_SPRINGBOOT---[topic]={}, [partition]={}, [offset]={}",
                            recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
                }
            });
        }
    }
}
